package raft

import (
	"sync"
	"time"
)

// -------------------------------------------------------------
// 
// Typedefs 
// 
// ------------------------------------

// /对等方是对协商一致协议中涉及的另一台服务器的引用。
type Peer struct {
	server            *server
	Name              string `json:"name"`
	ConnectionString  string `json:"connectionString"`
	prevLogIndex      uint64
	stopChan          chan bool
	heartbeatInterval time.Duration
	lastActivity      time.Time
	sync.RWMutex
}

// /-----------------------------------------------------------------
// 
// 构造函数
// 
// /----------------------------------------------------------------

// 创建一个新的对等点。
func newPeer(server *server, name string, connectionString string, heartbeatInterval time.Duration) *Peer {
	return &Peer{
		server:            server,
		Name:              name,
		ConnectionString:  connectionString,
		heartbeatInterval: heartbeatInterval,
	}
}

// ---------------------------------------------------------------
// 
// 访问器
// 
// ---------------------------------------------------------------

// 设置心跳超时。
func (p *Peer) setHeartbeatInterval(duration time.Duration) {
	p.heartbeatInterval = duration
}

// /---------------------------------------
// 上一个日志索引
// /---------------------------------------

// 检索上一个日志索引。
func (p *Peer) getPrevLogIndex() uint64 {
	p.RLock()
	defer p.RUnlock()
	return p.prevLogIndex
}

// 设置上一个日志索引。
func (p *Peer) setPrevLogIndex(value uint64) {
	p.Lock()
	defer p.Unlock()
	p.prevLogIndex = value
}

func (p *Peer) setLastActivity(now time.Time) {
	p.Lock()
	defer p.Unlock()
	p.lastActivity = now
}

// /------------------------------------------------------------------
// /
// 方法
// /
// /------------------------------------------------------------------

// /-----------------------------------------
// 心跳
// /---------------------------------------

// 启动对等心跳。
func (p *Peer) startHeartbeat() {
	p.stopChan = make(chan bool)
	c := make(chan bool)

	p.setLastActivity(time.Now())

	p.server.routineGroup.Add(1)
	go func() {
		defer p.server.routineGroup.Done()
		p.heartbeat(c)
	}()
	<-c
}

// 停止对等心跳。
func (p *Peer) stopHeartbeat(flush bool) {
	p.setLastActivity(time.Time{})

	p.stopChan <- flush
}

// LastActivity返回上次从对等方收到响应的时间。
func (p *Peer) LastActivity() time.Time {
	p.RLock()
	defer p.RUnlock()
	return p.lastActivity
}

// /-----------------------------------------
// 复制
// /------------------------------------

// 克隆对等方的状态。克隆未连接到服务器，
// 心跳计时器将不存在。
func (p *Peer) clone() *Peer {
	p.Lock()
	defer p.Unlock()
	return &Peer{
		Name:             p.Name,
		ConnectionString: p.ConnectionString,
		prevLogIndex:     p.prevLogIndex,
		lastActivity:     p.lastActivity,
	}
}

// /---------------------------------------
// 心跳
// /---------------------------------------

// 侦听心跳超时并刷新一个附加项。
func (p *Peer) heartbeat(c chan bool) {
	stopChan := p.stopChan

	c <- true

	ticker := time.Tick(p.heartbeatInterval)

	debugln("peer.heartbeat: ", p.Name, p.heartbeatInterval)

	for {
		select {
		case flush := <-stopChan:
			if flush {
				// 在安全删除节点之前
				// 我们必须先将删除命令刷新到节点
				p.flush()
				debugln("peer.heartbeat.stop.with.flush: ", p.Name)
				return
			} else {
				debugln("peer.heartbeat.stop: ", p.Name)
				return
			}

		case <-ticker:
			start := time.Now()
			p.flush()
			duration := time.Now().Sub(start)
			p.server.DispatchEvent(newEvent(HeartbeatEventType, duration, nil))
		}
	}
}

func (p *Peer) flush() {
	debugln("peer.heartbeat.flush: ", p.Name)
	prevLogIndex := p.getPrevLogIndex()
	term := p.server.currentTerm

	entries, prevLogTerm := p.server.log.getEntriesAfter(prevLogIndex, p.server.maxLogEntriesPerRequest)

	if entries != nil {
		p.sendAppendEntriesRequest(newAppendEntriesRequest(term, prevLogIndex, prevLogTerm, p.server.log.CommitIndex(), p.server.name, entries))
	} else {
		p.sendSnapshotRequest(newSnapshotRequest(p.server.name, p.server.snapshot))
	}
}

// ---------------------------------------
// 追加条目
// ---------------------------------------

// 向对等方发送追加条目请求通过交通工具。
func (p *Peer) sendAppendEntriesRequest(req *AppendEntriesRequest) {
	tracef("peer.append.send: %s->%s [prevLog:%v length: %v]\n",
		p.server.Name(), p.Name, req.PrevLogIndex, len(req.Entries))

	resp := p.server.Transporter().SendAppendEntriesRequest(p.server, p, req)
	if resp == nil {
		p.server.DispatchEvent(newEvent(HeartbeatIntervalEventType, p, nil))
		debugln("peer.append.timeout: ", p.server.Name(), "->", p.Name)
		return
	}
	traceln("peer.append.resp: ", p.server.Name(), "<-", p.Name)

	p.setLastActivity(time.Now())
	// 如果成功，则更新以前的日志索引。
	p.Lock()
	if resp.Success() {
		if len(req.Entries) > 0 {
			p.prevLogIndex = req.Entries[len(req.Entries)-1].GetIndex()

			// 如果peer追加当前术语
			// 中的日志项，我们将append设置为true 
			if req.Entries[len(req.Entries)-1].GetTerm() == p.server.currentTerm {
				resp.append = true
			}
		}
		traceln("peer.append.resp.success: ", p.Name, "; idx =", p.prevLogIndex)
		// 如果不成功，则减少上一个日志索引，
		// 我们下次再试。
	} else {
		if resp.Term() > p.server.Term() {
			// 当一个新的领导人出现时，这个*领导人*还不知道。
			// 此服务器可以知道，直到新领导发送一个带有更高术语
			// 的ae，或者此服务器完成此响应的处理。
			debugln("peer.append.resp.not.update: new.leader.found")
		} else if resp.Term() == req.Term && resp.CommitIndex() >= p.prevLogIndex {
			// 我们可能会错过对等方的响应
			// 因此，可能对等方提交了我们刚刚发送的日志
			// 但是我们没有收到成功的回复，也没有增加
			// PrevLoginEx 

			// 对等方此时未能截断日志并发送失败的回复
			// 我们只需要将对等方的prevLog索引更新为commitIndex 

			p.prevLogIndex = resp.CommitIndex()
			debugln("peer.append.resp.update: ", p.Name, "; idx =", p.prevLogIndex)

		} else if p.prevLogIndex > 0 {
			// 减小以前的日志索引，直到找到匹配项。不要
			// 让它位于对等方提交索引的下方。这是一个问题。
			p.prevLogIndex--
			// 如果不够，我们直接减少到
			if p.prevLogIndex > resp.Index() {
				p.prevLogIndex = resp.Index()
			}

			debugln("peer.append.resp.decrement: ", p.Name, "; idx =", p.prevLogIndex)
		}
	}
	p.Unlock()

	// 的索引，将对等点连接到resp，这样服务器就可以知道它来自哪里
	resp.peer = p.Name
	// 将响应发送到服务器进行处理。
	p.server.sendAsync(resp)
}

// 通过传输向对等方发送快照请求。
func (p *Peer) sendSnapshotRequest(req *SnapshotRequest) {
	debugln("peer.snap.send: ", p.Name)

	resp := p.server.Transporter().SendSnapshotRequest(p.server, p, req)
	if resp == nil {
		debugln("peer.snap.timeout: ", p.Name)
		return
	}

	debugln("peer.snap.recv: ", p.Name)

	// 如果成功，对等方应该已处于快照状态
	// 将快照发送给它！
	p.setLastActivity(time.Now())

	if resp.Success {
		p.sendSnapshotRecoveryRequest()
	} else {
		debugln("peer.snap.failed: ", p.Name)
		return
	}

}

// 通过传输向对等方发送快照恢复请求。
func (p *Peer) sendSnapshotRecoveryRequest() {
	req := newSnapshotRecoveryRequest(p.server.name, p.server.snapshot)
	debugln("peer.snap.recovery.send: ", p.Name)
	resp := p.server.Transporter().SendSnapshotRecoveryRequest(p.server, p, req)

	if resp == nil {
		debugln("peer.snap.recovery.timeout: ", p.Name)
		return
	}

	p.setLastActivity(time.Now())
	if resp.Success {
		p.prevLogIndex = req.LastIndex
	} else {
		debugln("peer.snap.recovery.failed: ", p.Name)
		return
	}

	p.server.sendAsync(resp)
}

// /-----------------------------------------
// 投票请求
// /------------------------------------

// 发送投票请求
func (p *Peer) sendVoteRequest(req *RequestVoteRequest, c chan *RequestVoteResponse) {
	debugln("peer.vote: ", p.server.Name(), "->", p.Name)
	req.peer = p
	if resp := p.server.Transporter().SendVoteRequest(p.server, p, req); resp != nil {
		debugln("peer.vote.recv: ", p.server.Name(), "<-", p.Name)
		p.setLastActivity(time.Now())
		resp.peer = p
		c <- resp
	} else {
		debugln("peer.vote.failed: ", p.server.Name(), "<-", p.Name)
	}
}
